package com.jiayou.server

import com.jiayou.Protocol
import com.jiayou.session.ClientSocketSession
import com.jiayou.tool.objectToByteArray
import com.sun.org.apache.xpath.internal.operations.Bool
import io.ktor.network.selector.*
import io.ktor.network.sockets.*
import io.ktor.utils.io.*
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.io.Serializable
import java.net.InetSocketAddress
import java.util.concurrent.locks.Condition
import java.util.concurrent.locks.LockSupport
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.thread
import kotlin.properties.Delegates
import kotlin.reflect.KProperty

/**
@author lishuang
@data   2021-04-23  16:40:20
@info   客户端session服务
 *  T  ->  实际传递数据类型
 *  T1 ->  认证身份类型
 *  T2 ->  认证凭证类型
 */
open class ClientSocketServer<T : Serializable, T1 : Serializable, T2 : Serializable>(
    private val inetSocketAddress: InetSocketAddress = InetSocketAddress("localhost", 9000),
    val clientSocketSession: ClientSocketSession<T, T1, T2>
) {
    companion object {
        val logger: Logger = LoggerFactory.getLogger(ClientSocketServer::class.java)
    }

    //启动标志
    @Volatile
    private var reConnection: Boolean = true

    //同步启动锁标志
    private var syncThread: Thread? = null


    private val reentrantLock = ReentrantLock()
    private lateinit var output: ByteWriteChannel
    private lateinit var input: ByteReadChannel
    private lateinit var socket: Socket
    private lateinit var receiveHandler: (Protocol<T, T1, T2>) -> Unit
    private val service = thread(start = false, name = "ktor-client") {
        while (true) {
            runBlocking {
                //wait函数等待远程服务的响应（挂起）
                //同级的CoroutineScope是又有后顺序的
                try {
                    coroutineScope {
                        socket = aSocket(ActorSelectorManager(Dispatchers.IO)).tcp().configure {
                            println(this)
                        }.connect(inetSocketAddress) {
                            this.keepAlive = true
                        }
                        output = socket.openWriteChannel(true)
                        input = socket.openReadChannel()
                    }
                } catch (e: Exception) {
                    e.printStackTrace()
                    return@runBlocking
                }


                //认证协程
                try {
                    coroutineScope {
                        onOpen(socket, output, input)
                    }
                } catch (e: Exception) {
                    e.printStackTrace()
                    return@runBlocking
                }


                //监听数据流的协程
                launch {
                    try {
                        while (true) {
                            if (!onMessage(socket, output, input)) break
                        }
                    } catch (e: Throwable) {
                        e.printStackTrace()
                        onError(socket)
                    }
                }
                //监听socket关闭的协程
                launch {
                    try {
                        onWaitClosed(socket)
                    } catch (e: Exception) {
                        e.printStackTrace()
                    }
                }
                if (syncThread != null)
                    LockSupport.unpark(syncThread)
            }
            Thread.sleep(1000)
        }
    }


    private suspend fun onOpen(socket: Socket, output: ByteWriteChannel, input: ByteReadChannel) =
        clientSocketSession.onOpen(socket, output, input)

    private suspend fun onMessage(socket: Socket, output: ByteWriteChannel, input: ByteReadChannel) =
        clientSocketSession.onMessage(socket, output, input, ::onRecMessage)

    private suspend fun onWaitClosed(socket: Socket) =
        clientSocketSession.onWaitClosed(socket)

    private fun onError(socket: Socket) =
        clientSocketSession.onError(socket)

    private fun reConnection(reconnection: Boolean) {
        reConnection = reconnection
    }

    fun tryToConnection() {
        try {
            if (service.isAlive) {
                LockSupport.unpark(service)
            }
        } catch (e: Exception) {
            logger.error("重连操作失败，原因：")
            e.printStackTrace()
        }
    }

    @Synchronized
    fun start(async: Boolean = true) {
        syncThread = Thread.currentThread()
        thread {
            if (!service.isAlive) {
                service.start()
            }
        }
        if (!async) {
            LockSupport.park()
        }
        //help gc
        syncThread = null
        logger.info("ktor client connection to ${inetSocketAddress.hostName}:${inetSocketAddress.port}")
    }

    @Synchronized
    fun stop() {
        try {
            if (!socket.isClosed)
                socket.close()
        } catch (e: InterruptedException) {
            logger.info("ktor client stoped!")
        } catch (e: Exception) {
            e.printStackTrace()
        }

    }

    private fun onRecMessage(protocol: Protocol<T, T1, T2>) {
        if (this::receiveHandler.isInitialized) {
            reentrantLock.lock()
            try {
                this.receiveHandler.invoke(protocol)
            } catch (e: Exception) {
                e.printStackTrace()
            } finally {
                reentrantLock.unlock()
            }
        } else {
            logger.info("rec message : $protocol")
        }
    }

    fun setReceiveHandler(receiveHandler: (Protocol<T, T1, T2>) -> Unit) {
        reentrantLock.lock()
        try {
            this.receiveHandler = receiveHandler
        } catch (e: Exception) {
            e.printStackTrace()
        } finally {
            reentrantLock.unlock()
        }
    }


    @Synchronized
    fun sendMessage(protocol: Protocol<T, T1, T2>) {
        val objectToByteArray = objectToByteArray(protocol)
        runBlocking {
            launch {
                output.writeInt(objectToByteArray.size)
                output.writeFully(objectToByteArray)
                output.flush()
            }
        }
    }

}